package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/2/5
 * 
 *  使用TableAPI 或 SQL编程，需要有编程模型 Table（动态表）
 *          动态表： 表会随着数据而变化
 *          
 *     针对动态表，发起的查询会随着数据的到来，不断查询，连续查询。
 */
public class Demo1_Table
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //先有Table的编程环境
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //把流转换为动态表
        Table table = tableEnvironment.fromDataStream(ds);

        //打印表的结构 取决于流中的类型的属性
        table.printSchema();

        /*
                连续查询
                示例:
                tab.select($("key"), $("value").avg().plus(" The average").as("average"));

                select(Expression x)
                    Expression: 代表可以产生计算结果的计算逻辑。
                                Expression可以包含多个子表达式。
                                Expression可以是字面量，或字段的引用，或调用函数等
                                可以通过Expressions 构造，Expressions中提供了很多现成的方法，返回Expression。

                            举例:  Expressions.$(字段名)，选取某个字段的值。
         */
        Table result = table.select($("id"), $("ts"), $("vc"), $("vc").plus(10).as("vc+10"))
                           .where($("id").isEqual("s1"));

        //直接使用表
        result.execute().print();


        /*try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }*/
        
    }
}
